package com.streaming.project

import com.streaming.project.spark.dao.{CourseClickCountDAO, CourseSearchClickCountDAO}
import com.streaming.project.spark.domain.{ClickLog, CourseClickCount, CourseSearchClickCount}
import com.streaming.project.spark.utils.DateUtils
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

object StatStreamingApp {

  def main(args: Array[String]): Unit = {

    if(args.length != 4) {

      println("Usage:  you nedd to set values<ssc,zkQuorm,groupId,topicMap>")

      System.exit(1)

    }

    // 创建数组，包含zk, goupId, topics,线程数

    val Array(zkQuorm,groupId,topics,numThread) = args

    // 创建sparkconf

    val sparkConf = new SparkConf().setAppName("StatStreamingApp")//.setMaster("local[5]")

    // 创建SteamingContext,这个类包含两个参数，一个是SparkConf实例，另一个是interval

    val ssc = new StreamingContext(sparkConf,Seconds(60))

    //由于createStream这个函数传递的参数分别是StreamingContext,String,String,Map

    //val topicMap1 = topics.split(",").map((_,numThread.toInt)).toMap

    val topicMap = topics.split(",").map((_,numThread.toInt)).toMap

    // KafkaUtils 在2.3 2.4中过时了，但是可以用

    // get messages

    val messages = KafkaUtils.createStream(ssc,zkQuorm,groupId,topicMap)

    // step 1 测试数据接收

    //messages.print(100)

    //messages.map(_._2).count().print

    //98.167.187.30	2019-03-02 17:43:00	"GET /class/145.html HTTP/1.1"	300

    // 数据清洗c

    val logs = messages.map(_._2)

    val logs2 = messages.map(x=>x.toString().split(",")(1))

    //println("第二格式")

    //logs2.print(100)

    //println("第一种格式")

    //logs.print(100)

    val clearnData = logs.map(line=>{


      val infos = line.split("\t")

      val ip = infos(0)

      val time = infos(1)+infos(2)

      // infos(2) :"GET /class/146.html HTTP/1.1"

      // /class/146.html

      val url = infos(2).split(" ")(1)

      var courseId = 0

      // 拿到课程编号

      if(url.startsWith("/class")){

        val courseHTML = url.split("/")(2)

        courseId = courseHTML.substring(0,courseHTML.lastIndexOf(".")).toInt

      }

      ClickLog(infos(0),DateUtils.parseToMinute(infos(1)),courseId,infos(3).toInt,infos(4))

    }).filter(ClickLog=>ClickLog.courseId!=0)

    clearnData.print()
    //clearnData.count().print()

    // 统计今天到现在为止实战课程的访问量，功能1 yyyyMMdd courseId

  clearnData.map(x=>{

    // the design of Rowkey

    (x.time.substring(0,8)+"_"+x.courseId,1)

  }).reduceByKey(_+_).foreachRDD(rdd=>{

    rdd.foreachPartition(partitionRecord=>{

      val list = new ListBuffer[CourseClickCount]

      partitionRecord.foreach(pair=>{

        list.append(CourseClickCount(pair._1,pair._2))

      })

      CourseClickCountDAO.sava(list)

    })

  })

    //测试四：统计从搜索引擎过来的今天到现在为止的实战课程的访问量
    // x.referer ==>https://www.sogou.com/web?query=Spark_Streaming实战
    // x.referer.replace("//","/")==>https:/www.sogou.com/web?query=Spark_Streaming实战

    clearnData.map(x=>{

      val referer = x.referer.replace("//","/")

      val splits = referer.split("/")

      var host= ""

      if(splits.length>2){

        host = splits(1) // www.sogou.com

      }

      (host,x.courseId,x.time) // www.sogou.com,12,20190303155600


    }).filter(_._1!="")//当courseId 为 null 的过滤掉
      .map(e=>{

      (e._3.substring(0,8)+"_"+e._1+"_"+e._2,1)

      //==>(20190303_www.sogou.com_145,1)

    }).reduceByKey(_+_).foreachRDD(rdd=>{

      rdd.foreachPartition(partitionRecords=>{

        val list = new ListBuffer[CourseSearchClickCount]

        partitionRecords.foreach(pair=>{

          list.append(CourseSearchClickCount(pair._1,pair._2))

        })

        CourseSearchClickCountDAO.save(list)

      })

    })

    clearnData.map(x=>{

      val referer = x.referer.replace("//","/")

      val splits = referer.split("/")

      var host= ""

      if(splits.length>2){

        host = splits(1) // www.sogou.com

      }

      (host,x.courseId,x.time) // www.sogou.com,12i,20190303155600


    }).filter(_._1!="")

    ssc.start()
    ssc.awaitTermination()

  }


}
